package com.relytosoft.mizar.analyze.strategy.consumer;

import com.alibaba.fastjson.JSON;
import com.relytosoft.dubhe.common.core.constant.CommonConstants;
import com.relytosoft.dubhe.common.core.object.ResponseResult;
import com.relytosoft.dubhe.common.kafka.producer.KafkaOpMsg;
import com.relytosoft.dubhe.common.kafka.producer.KafkaProducer;
import com.relytosoft.mizar.analyze.base.constant.MediaMqConstants;
import com.relytosoft.mizar.analyze.base.entity.pojo.dto.FaceGroupReqDto;
import com.relytosoft.mizar.analyze.base.entity.pojo.dto.FacePersonReqDto;
import com.relytosoft.mizar.analyze.strategy.algorithm.face.service.FaceService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * PersonConsumer
 *
 * @Author SunLi
 * @Date: 2022/5/16 19:41
 * @Version 1.0
 */
@Component
public class PersonConsumer {
    protected Logger log = LoggerFactory.getLogger(this.getClass());

    @Resource
    private FaceService faceService;

    @Resource
    private KafkaProducer kafkaProducer;

    @KafkaListener(id = "consumer-analyze-face-person", topics = MediaMqConstants.FACE_ANALYZE_PERSON_TOPIC, containerFactory = "singleContainerFactory")
    public void processSingleMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        try {
            boolean isConsume = false;
            KafkaOpMsg kafkaOpMsg = JSON.parseObject(record.value().toString(), KafkaOpMsg.class);
            String data = kafkaOpMsg.getData();
            switch (kafkaOpMsg.getOpType()) {
                case "insertGroup":
                    isConsume = createGroup(data);
                    break;
                case "deleteGroup":
                    isConsume = deleteGroup(data);
                    break;
                case "insertPerson":
                    isConsume = createPerson(data);
                    break;
                case "deletePerson":
                    isConsume = deletePerson(data);
                    break;
                default:
                    break;
            }
            log.info("线程:{} 消费-人脸库及人脸同步->{}", Thread.currentThread(), data);
            if (isConsume) {
                ack.acknowledge();
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    /**
     * 人脸库添加
     *
     * @param data 人脸库信息
     * @return 消费结果
     */
    private boolean createGroup(String data) {
        FaceGroupReqDto faceGroupReqDto = JSON.parseObject(data, FaceGroupReqDto.class);
        ResponseResult result = faceService.createGroup(faceGroupReqDto);
        if (result.getCode() == CommonConstants.SUCCESS) {
            KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
            kafkaOpMsg.setData(JSON.toJSONString(result.getData()));
            kafkaProducer.send(MediaMqConstants.FACE_ANALYZE_GROUP_ID_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
            return true;
        }
        return false;
    }

    /**
     * 人脸库删除
     *
     * @param data 人脸库信息
     * @return 消费结果
     */
    private boolean deleteGroup(String data) {
        FaceGroupReqDto faceGroupReqDto = JSON.parseObject(data, FaceGroupReqDto.class);
        ResponseResult result = faceService.deleteGroup(faceGroupReqDto);
        if (result.getCode() == CommonConstants.SUCCESS) {
            return true;
        }
        return false;
    }

    /**
     * 人脸添加
     *
     * @param data 人脸信息
     * @return 消费结果
     */
    private boolean createPerson(String data) {
        FacePersonReqDto facePersonReqDto = JSON.parseObject(data, FacePersonReqDto.class);
        ResponseResult result = faceService.createPerson(facePersonReqDto);
        if (result.getCode() == CommonConstants.SUCCESS) {
            KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
            kafkaOpMsg.setData(JSON.toJSONString(result.getData()));
            kafkaProducer.send(MediaMqConstants.FACE_ANALYZE_PERSON_ID_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
            return true;
        }
        return false;
    }

    /**
     * 人脸删除
     *
     * @param data 人脸信息
     * @return 消费结果
     */
    private boolean deletePerson(String data) {
        FacePersonReqDto facePersonReqDto = JSON.parseObject(data, FacePersonReqDto.class);
        ResponseResult result = faceService.deletePerson(facePersonReqDto);
        if (result.getCode() == CommonConstants.SUCCESS) {
            return true;
        }
        return false;
    }
}
